package com.qilin.hscs.hm.mqtt.handler;

import com.qilin.hscs.hm.mqtt.constant.HostNameConstant;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;


/**
 * Created by 10188 on 2018/3/23.
 */
@Component
public class OutterHostRepeater extends Repeater{
    
    private static final Logger logger= LogManager.getLogger(OutterHostRepeater.class.getName());

    private MqttClient clienta;

    private MqttClient repeaterClient;

    private String clientIda = "outterClienta";

    private String repeaterClientId="outterRepeaterClient";

    @Autowired
    @Qualifier(value = "outterHost")
    private  String  host;


    @Override
    public void repeat() throws MqttException {
        repeaterClient = factory.getClient(repeaterClientId, host);
        casUtil.send(repeaterClient,msg,topicVar);
    }

    @Override
    @PostConstruct
    public void listenFromApp() throws Exception {
        logger.info("outterClient启动");
        clienta = factory.getClient(clientIda, host);
        casUtil.receive(clienta, topicFromApp, new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {

            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                msg = new String(mqttMessage.getPayload());
                topicVar=s;
                logger.info("【外部转发器收到app端消息】={}",msg);
                flushToHandler();
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

            }
        });

    }

    @Override
    void flushToHandler() {
        Observable<String> observable= Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext(HostNameConstant.outterHostName+"#"+topicVar+"#"+msg);
            }
        });
        observable.subscribe(handler);
    }


    @Override
    public void onSubscribe(Disposable disposable) {

    }

    @Override
    public void onNext(String s) {
        logger.info("【收到处理中心发送的消息】={}",s);
        String[] content = s.split("#");
        String hostName = content[0];
        topicVar = content[1];
        msg = content[2];
        try {
            repeat();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onError(Throwable throwable) {

    }

    @Override
    public void onComplete() {

    }
}
